 1.Spark Streaming之与Kafka整合中Offset 管理
   
   Spark Streaming集成Kafka，允许从Kafka中读取一个或者多个 topic 的数据。一个
Kafka Topic包含一个或多个分区，每个分区中的消息顺序存储，并使用 offset 来标
记消息的位置。开发者可以在 Spark Streaming 应用中通过 offset 来控制数据的读
取位置。
   Offsets 管理对于保证流式应用在整个生命周期中数据的连贯性是非常重要的。如果
在应用停止或报错退出之前没有将 offset 持久化保存，该信息就会丢失，那么Spark
Streaming就没有办法从上次停止或报错的位置继续消费Kafka中的消息。
   1).获取偏移量(Obtaining Offsets)
   Spark Streaming与kafka整合时，允许获取其消费的 offset ，具体方法如下：
stream.foreachRDD { rdd =>
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  rdd.foreachPartition { iter =>
    val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
    println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
  }
}  
   注意：对HasOffsetRanges的类型转换只有在对 createDirectStream 调用的第一个
方法中完成时才会成功，而不是在随后的方法链中。RDD分区和Kafka分区之间的对应关
系在 shuffle 或 重分区后会丧失，如reduceByKey 或 window。
   2).存储偏移量(Storing Offsets)
   在Streaming程序失败的情况下，Kafka交付语义取决于如何以及何时存储偏移量。
Spark输出操作的语义为 at-least-once。
   如果要实现EOS语义(Exactly Once Semantics)，必须在幂等的输出之后存储偏移量
或者 将存储偏移量与输出放在一个事务中。可以按照增加可靠性(和代码复杂度)的顺
序使用以下选项来存储偏移量：
   Checkpoint
   Checkpoint是对Spark Streaming运行过程中的元数据和每RDDs的数据状态保存到一
个持久化系统中，当然这里面也包含了offset，一般是HDFS、S3，如果应用程序或集群
挂了，可以迅速恢复。
   如果Streaming程序的代码变了，重新打包执行就会出现反序列化异常的问题。
   这是因为Checkpoint首次持久化时会将整个 jar 包序列化，以便重启时恢复。重新打
包之后，新旧代码逻辑不同，就会报错或仍然执行旧版代码。
   要解决这个问题，只能将HDFS上的checkpoint文件删除，但这样也会同时删除Kafka的 
offset信息。
   Kafka
   默认情况下，消费者定期自动提交偏移量，它将偏移量存储在一个特殊的Kafka主题中
(__consumer_offsets)。但在某些情况下，这将导致问题，因为消息可能已经被消费者从
Kafka拉去出来，但是还没被处理。
   可以将 enable.auto.commit 设置为 false ，在 Streaming 程序输出结果之后，手动
提交偏移到kafka。
   与检查点相比，使用Kafka保存偏移量的优点是无论应用程序代码如何更改，偏移量
仍然有效。
stream.foreachRDD { rdd =>
  val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
  // 在输出操作完成之后，手工提交偏移量；此时将偏移量提交到 Kafka 的消息队列中
  stream.asInstanceOf[CanCommitOffsets].commitAsync(offsetRanges)
}  
   
   自定义存储
   Offsets可以通过多种方式来管理，但是一般来说遵循下面的步骤:
       在 DStream 初始化的时候，需要指定每个分区的offset用于从指定位置读取数据
       读取并处理消息
       处理完之后存储结果数据
	   用虚线圈存储和提交offset，强调用户可能会执行一系列操作来满足他们更加严
格的语义要求。这包括幂等操作和通过原子操作的方式存储offset
       将 offsets 保存在外部持久化数据库如 HBase、Kafka、HDFS、ZooKeeper、
Redis、MySQL ... ...
   
   可以将 Offsets 存储到HDFS中，但这并不是一个好的方案。因为HDFS延迟有点高，
此外将每批次数据的offset存储到HDFS中还会带来小文件问题；
   可以将 Offset 存储到保存ZK中，但是将ZK作为存储用，也并不是一个明智的选择，
同时ZK也不适合频繁的读写操作；
   3).Redis管理的Offset
   要想将Offset保存到外部存储中，关键要实现以下几个功能：
       Streaming程序启动时，从外部存储获取保存的Offsets(执行一次)
       在foreachRDD中，每个批次数据处理之后，更新外部存储的offsets(多次执行)
   Redis管理的Offsets：
   (1)、数据结构选择：Hash；key、field、value
        Key：kafka:topic:TopicName:groupid
        Field：partition
        Value：offset
   (2)、从 Redis 中获取保存的offsets
   (3)、消费数据后将offsets保存到redis
   案例一：使用自定义的offsets，从kafka读数据；处理完数据后打印offsets
package cn.lagou.sparksql.kafka

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, TaskContext}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{
  ConsumerStrategies,
  HasOffsetRanges, KafkaUtils, LocationStrategies, OffsetRange
}
import org.apache.spark.streaming.{Seconds, StreamingContext}

// 使用自定义的offsets，从kafka读数据；处理完数据后打印offsets
object KafkaDStream2 {
  def main(args: Array[String]): Unit = {
    // 初始化
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf: SparkConf = new SparkConf()
      .setMaster("local[*]")
      .setAppName(this.getClass.getCanonicalName)
    val ssc = new StreamingContext(conf, Seconds(2))

    // 定义kafka相关参数
    val groupId: String = "mygroup01"
    val topics: Array[String] = Array("topicB")
    val kafkaParams: Map[String, Object] = getKafkaConsumerParams(groupId)

    // 从指定的位置获取kafka数据
    val offsets: collection.Map[TopicPartition, Long] = Map(
      new TopicPartition("topicB", 0) -> 1000,
      new TopicPartition("topicB", 1) -> 1000,
      new TopicPartition("topicB", 2) -> 1000
    )
    // 从 kafka 中获取数据
    val dstream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream[String, String](
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsets)
      )
    // 创建DStream
    dstream.foreachRDD { (rdd, time) =>
      // 输出结果
      println(s"*********** rdd.count = ${rdd.count()}; time = $time ***********")

      // 输出offset
      val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

      rdd.foreachPartition { iter =>
        val range: OffsetRange = offsetRanges(TaskContext.get.partitionId)
        // 输出kafka消费的offset
        println(s"${range.topic} ${range.partition} ${range.fromOffset} ${range.untilOffset}")
      }
    }

    ssc.start()
    ssc.awaitTermination()
  }

  def getKafkaConsumerParams(groupId: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux121:9092,linux122:9092,linux123:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupId,
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean))
  }
}

   
   // kafka命令，检查 topic offset的值
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list
linux121:9092,linux122:9092,linux123:9092 --topic topicB --time -1
   
   案例二：根据 key 从 Redis 获取offsets，根据该offsets从kafka读数据；处理完数据
后将offsets保存到 Redis
   
   引入依赖
<!-- jedis -->
<dependency>
    <groupId>redis.clients</groupId>
    <artifactId>jedis</artifactId>
    <version>2.9.0</version>
</dependency>
   主程序(从kafka获取数据，使用 Redis 保存offsets)
package cn.lagou.Streaming.kafka

import org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.kafka.common.TopicPartition
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.log4j.{Level, Logger}
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.{SparkConf, TaskContext}

object KafkaDStream3 {
  def main(args: Array[String]): Unit = {
    // 初始化
    Logger.getLogger("org").setLevel(Level.ERROR)
    val conf: SparkConf = new SparkConf().setAppName(this.getClass.getCanonicalName)
      .setMaster("local[*]")
    val ssc = new StreamingContext(conf, Seconds(5))

    // 定义kafka相关参数
    val groupId: String = "lagou_group01"
    val topics: Array[String] = Array("lagou_topic01", "lagou_topic02")
    val kafkaParams: Map[String, Object] = getKafkaConsumerParams(groupId)

    // 从redis中获取offsets
    val offsets: Map[TopicPartition, Long] =
      OffsetsWithRedisUtils.getOffsetsFromRedis(topics, groupId)
    // 创建DStream
    val dstream: InputDStream[ConsumerRecord[String, String]] =
      KafkaUtils.createDirectStream(
        ssc,
        LocationStrategies.PreferConsistent,
        ConsumerStrategies.Subscribe[String, String](topics, kafkaParams, offsets)
      )

    // DStream转换&输出
    dstream.foreachRDD { (rdd, time) =>
      if (!rdd.isEmpty()) {
        // 处理消息
        println(s"************ rdd.count = ${rdd.count()}; " +
          s"time =$time*********")
        // 将offsets信息打印到控制台
        val offsetRanges: Array[OffsetRange] = rdd.asInstanceOf[HasOffsetRanges].offsetRanges

        rdd.foreachPartition { iter =>
          val range: OffsetRange = offsetRanges(TaskContext.get.partitionId)
          // 输出kafka消费的offset
          println(s"${range.topic} ${range.partition} ${range.fromOffset} ${range.untilOffset}")
        }

        // 将offsets保存到redis
        OffsetsWithRedisUtils.saveOffsetsToRedis(offsetRanges, groupId)
      }
    }

    // 启动作业
    ssc.start()
    ssc.awaitTermination()
  }

  def getKafkaConsumerParams(groupid: String): Map[String, Object] = {
    Map[String, Object](
      ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux121:9092,linux122:9092,linux123:9092",
      ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -> classOf[StringDeserializer],
      ConsumerConfig.GROUP_ID_CONFIG -> groupid,
      ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -> (false: java.lang.Boolean)
      //ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -> "earliest"
    )
  }
}

// kafka-topics.sh --zookeeper linux121:2181,linux122:2181,linux123:2181/myKafka
// --create --topic lagou_topic01 --replication-factor 2 --partitions 3
// kafka-topics.sh --zookeeper linux121:2181,linux122:2181,linux123:2181/myKafka
// --create --topic lagou_topic02 --replication-factor 2 --partitions 2
   
   工具类(Redis读取/保存offsets)
package cn.lagou.Streaming.kafka

import java.util
import org.apache.kafka.common.TopicPartition
import org.apache.spark.streaming.kafka010.OffsetRange
import redis.clients.jedis.{Jedis, JedisPool, JedisPoolConfig}

import scala.collection.mutable

object OffsetsWithRedisUtils {
  // 定义redis的参数
  private val redisHost = "192.168.71.100"
  private val redisPort = 6379

  // 获取redis的连接
  private val config = new JedisPoolConfig
  // 最多空闲数
  config.setMaxIdle(5)
  // 最大连接数
  config.setMaxTotal(10)

  private val poll = new JedisPool(config, redisHost, redisPort, 10000)
  private def getRedisConnection:Jedis =poll.getResource

  private val topicPrefix = "kafka:topic"

  // key: kafka:topic:TopicName:groupid
  private def getKey(topic: String, groupid: String) =s"$topicPrefix:$topic:$groupid"

  // 根据key获取offsets
  def getOffsetsFromRedis(topics: Array[String], groupId: String): Map[TopicPartition, Long]= {
    val jedis: Jedis = getRedisConnection

    val offsets: Array[mutable.Map[TopicPartition, Long]] = topics.map { topic =>
      val key = getKey(topic, groupId)
      import scala.collection.JavaConverters._

      jedis.hgetAll(key)
        .asScala
        .map { case (partition, offset) => new TopicPartition(topic, partition.toInt) -> offset.toLong }
    }
    // 归还资源
    jedis.close()

    offsets.flatten.toMap
  }

  // 将offsets保存到redis中
  def saveOffsetsToRedis(offsets: Array[OffsetRange], groupId: String): Unit={
    // 获取连接
    val jedis: Jedis = getRedisConnection

    // 组织数据
    offsets.map{rang => (rang.topic, (rang.partition.toString, rang.untilOffset.toString))}
        .groupBy(_._1)
        .foreach{case (topic, buffer) =>
          val key = getKey(topic, groupId)
          import scala.collection.JavaConverters._
          val maps: util.Map[String, String] = buffer.map(_._2).toMap.asJava
          // 保存数据 key Map[String, String](partition, offset)
          jedis.hmset(key, maps)

        }
    jedis.close()
  }

  def main(args: Array[String]): Unit = {
    val groupid = "group01"
    val topic = "topicB"
    val offsets = Array(
      OffsetRange(topic, 0,   0,  600),
      OffsetRange(topic, 1, 100, 1000),
      OffsetRange(topic, 2, 200, 2000)
    )
    saveOffsetsToRedis(offsets, groupid)

    val groupId: String = "group01"
    val topics: Array[String] = Array("topicB")
    val offsets1: Map[TopicPartition, Long] = getOffsetsFromRedis(topics, groupId)
    offsets1.foreach(println)
  }
}

// hmset kafka:topic:topicB:group01 0 0 1 100 2 200
   
   